/**
 * Copyright (c) 2013-2019 Nikita Koksharov
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.redisson.pubsub;

import org.redisson.RedissonLockEntry;
import org.redisson.misc.RPromise;

/**
 * 
 * @author Nikita Koksharov
 *
 */
public class LockPubSub extends PublishSubscribe<RedissonLockEntry> {
    /**
     * 解锁消息
     */
    public static final Long UNLOCK_MESSAGE = 0L;
    /**
     * 读已解锁消息。只会在RedissonWriteLock中使用到
     */
    public static final Long READ_UNLOCK_MESSAGE = 1L;

    public LockPubSub(PublishSubscribeService service) {
        super(service);
    }
    
    @Override
    protected RedissonLockEntry createEntry(RPromise<RedissonLockEntry> newPromise) {
        return new RedissonLockEntry(newPromise);
    }

    /**
     * 当redis向通道A发送消息时，监听通道A的监听器(与一个RedissonLockEntry绑定)调用该方法进行处理。
     *
     * @param value RedissonLockEntry
     * @param message 消息
     */
    @Override
    protected void onMessage(RedissonLockEntry value, Long message) {
        // 非读写锁的解锁消息
        if (message.equals(UNLOCK_MESSAGE)) {
            Runnable runnableToExecute = value.getListeners().poll();
            if (runnableToExecute != null) {
                runnableToExecute.run();
            }

            // 核心。将该通道对应的RedissonLockEntry.latch释放一个信号量。
            // 于是所有正在等待RedissonLockEntry.latch.tryAcquire()的线程立刻开始抢这一个信号量，谁抢到谁去获取锁
            value.getLatch().release();

        // 读写锁的读已解锁消息
        } else if (message.equals(READ_UNLOCK_MESSAGE)) {
            while (true) {
                Runnable runnableToExecute = value.getListeners().poll();
                if (runnableToExecute == null) {
                    break;
                }
                runnableToExecute.run();
            }

            // 核心。将该通道对应的RedissonLockEntry.latch释放等于当前所有尝试获取读锁的线程的数量
            // 因为读锁是不阻塞的，因此所有尝试获取读锁的线程都会去获取锁
            value.getLatch().release(value.getLatch().getQueueLength());
        }
    }

}
